package com.caption.mq.core;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.caption.mq.constant.MessageExtConstant;
import com.caption.mq.entity.MessageDTO;
import com.caption.mq.entity.MqSubscribeDO;
import com.caption.mq.handler.MessageHandler;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @description: TODO
 * @author: 码农
 * @date: 2022-01-16 06:34:30
 */
@Slf4j
public class RocketMqConsumer extends AbstractMqConsumer {

    private DefaultMQPushConsumer mqPushConsumer;
    @Setter
    @Getter
    private int retryCount = 3;
    /**
     * 消息消费失败默认的重试16次，但这16次其中是有间隔的。1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * 每一次重试都会递增时间，如果可以通过 ConsumeConcurrentlyContext.setDelayLevelWhenNextConsume(3) 设置重试间隔。注意这个3指的是 10s
     */
    @Setter
    @Getter
    private int delayLevelWhenNextConsume = 0;

    @Setter
    @Getter
    private Boolean started;


    public RocketMqConsumer(String nameServerAddress, MqSubscribeDO subscribe) {
        super(subscribe);
        this.retryCount = subscribe.getRetryCount();
        /**
         * 重试策略设置大于0时, 消息消费失败时重试时间从5s之后重新投递
         */
        if (subscribe.getRetryStrategy() > 0) {
            this.delayLevelWhenNextConsume = 2;
        } else {
            this.delayLevelWhenNextConsume = 0;
        }

        log.info("mqSubscribe配置{}", JSONObject.toJSONString(subscribe));
        init(nameServerAddress, subscribe);
    }

    public void init(String nameServerAddress, MqSubscribeDO subscribe) {
        mqPushConsumer = new DefaultMQPushConsumer(subscribe.getConsumeGroup());
        mqPushConsumer.setNamesrvAddr(nameServerAddress);
        mqPushConsumer.setMessageModel(MessageModel.valueOf(subscribe.getMessageMode()));
        mqPushConsumer.setConsumeThreadMax(subscribe.getConsumeThreadMax());
        mqPushConsumer.setConsumeThreadMin(subscribe.getConsumeThreadMin());
        mqPushConsumer.setPullBatchSize(subscribe.getPullBatchSize());
        mqPushConsumer.setConsumeMessageBatchMaxSize(subscribe.getConsumeMessageBatchMaxSize());
        //消息消费失败重试次数，超过重试次数会被转到死信队列
        mqPushConsumer.setMaxReconsumeTimes(subscribe.getRetryCount());
    }


    @Override
    public void subscribe(String topic, String tag) {
        try {
            mqPushConsumer.subscribe(topic, tag);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void registerMessageHandler(MessageHandler messageHandler) {
        if (MessageExtConstant.CONSUME_MODE_CONCURRENTLY.equals(subscribe.getConsumeMode())) {
            mqPushConsumer.registerMessageListener(new DefaultMessageListenerConcurrently(messageHandler));
        } else if (MessageExtConstant.CONSUME_MODE_ORDERLY.equals(subscribe.getConsumeMode())) {
            mqPushConsumer.registerMessageListener(new DefaultMessageListenerOrderly(messageHandler));
        } else {
            throw new RuntimeException("unknown consume mode ! only support CONCURRENTLY and ORDERLY");
        }
    }

    @Override
    public void start() {
        try {
            mqPushConsumer.start();
            this.setStarted(true);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void stop() {
        mqPushConsumer.unsubscribe(this.subscribe.getTopic());
        mqPushConsumer.shutdown();
        this.setStarted(false);
    }

    /**
     * 普通消息
     */
    public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {

        private MessageHandler messageHandler;

        public DefaultMessageListenerConcurrently(MessageHandler messageHandler) {
            super();
            this.messageHandler = messageHandler;
        }

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            MessageDTO messageDTO = null;
            for (MessageExt messageExt : msgs) {
                try {
                    messageDTO = getMessageDTO(messageExt);
                    messageHandler.onMessage(messageDTO);
                } catch (Exception e) {
                    //出现异常请重试a
                    log.error("消费异常",e);
                    return dealFailed(context, messageExt, messageDTO, e);
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        private ConsumeConcurrentlyStatus dealFailed(ConsumeConcurrentlyContext context, MessageExt messageExt, MessageDTO messageDTO, Exception e) {
            //重试次数没有超过限制
            int reconTimes = messageExt.getReconsumeTimes();
            if (reconTimes < retryCount) {
                context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
            } else {
                messageHandler.onMessageHandleFailed(messageDTO, e);
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }


    /**
     * 顺序消息，会一直等待前面的消息执行成功才会继续执行
     */
    public class DefaultMessageListenerOrderly implements MessageListenerOrderly {

        private MessageHandler messageHandler;

        public DefaultMessageListenerOrderly(MessageHandler messageHandler) {
            super();
            this.messageHandler = messageHandler;
        }

        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            for (MessageExt messageExt : msgs) {
                MessageDTO messageDTO = null;
                try {
                    messageDTO = getMessageDTO(messageExt);
                    messageHandler.onMessage(messageDTO);

                } catch (Exception e) {
                    context.setSuspendCurrentQueueTimeMillis(10);
                    messageHandler.onMessageHandleFailed(messageDTO, e);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
            return ConsumeOrderlyStatus.SUCCESS;
        }
    }


    /**
     * 解析消息体
     * @param messageExt
     * @return
     */
    private MessageDTO getMessageDTO(MessageExt messageExt) throws UnsupportedEncodingException {
        log.info("接收到消息messageExt{}",messageExt);
        String result = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
        MessageDTO messageDTO = JSON.parseObject(result, MessageDTO.class);
        messageDTO.setMsgId(messageExt.getMsgId());
        return messageDTO;
    }
}
